// use polars::{df, prelude::*};
// use rand::{distr::Alphabetic, seq::IndexedRandom, Rng};

// fn main(){
//  let mut start_day=chrono::NaiveDate::from_ymd_opt(2020, 1, 1).unwrap().and_hms_opt(0, 0, 0).unwrap();
//  let end_day=chrono::NaiveDate::from_ymd_opt(2023, 1, 1).unwrap().and_hms_opt(0, 0, 0).unwrap();
//  while start_day<end_day {
//      write(&mut start_day);
//  }
// }

// fn write(start_day:&mut chrono::NaiveDateTime){
//     let mut dates=Vec::new();
//     let path=format!("data/{}.par",start_day.format("%Y-%m-%d").to_string());
//     let end_day=start_day.checked_add_days(chrono::Days::new(1)).unwrap();
//     while start_day.to_owned()<end_day {
//         dates.push(start_day.clone());
//         *start_day=start_day.checked_add_signed(chrono::TimeDelta::hours(1)).unwrap();
//     }
//     let choice=["一分","二分","三分"];
//     let capacity=1000_000_0;
//     let mut aa=Vec::with_capacity(capacity);
//     let mut bb=Vec::with_capacity(capacity);
//     let mut cc=Vec::with_capacity(capacity);
//     let mut dd=Vec::with_capacity(capacity);
//     let mut ee=Vec::with_capacity(capacity);
//     let mut rng=rand::rng();
//     for _ in 0..capacity {
//         aa.push(choice.choose(&mut rng).unwrap().to_string());
//         bb.push(rng.clone().sample_iter(Alphabetic).take(2).map(char::from).collect::<String>());
//         cc.push(dates.choose(&mut rng).unwrap().to_owned());
//         dd.push(rng.clone().sample_iter(Alphabetic).take(1).map(char::from).collect::<String>());
//         ee.push(rng.random_range(0..100u32));
//     }
//     let mut df=df!("dan_wei"=>aa,"saler"=>bb,"sale_time"=>cc,"sale_good"=>dd,"sale_num"=>ee).unwrap();
//     let f=std::fs::File::create(path).unwrap();
//     ParquetWriter::new(f).finish(&mut df).unwrap();
// }

// #[cfg(test)]
// mod t{
//     use polars::prelude::{col, lit, LazyFrame, ScanArgsParquet};

//     #[test]
//     fn tt() {
//         let paths=std::fs::read_dir("../data").unwrap();
//         let paths=paths.into_iter().map(|x|x.unwrap().path()).collect();
//         let df=LazyFrame::scan_parquet_files(paths, ScanArgsParquet::default()).unwrap()
//         .filter(col("sale_time").dt().strftime("%Y-%m").eq(lit("2024-01")));
//         println!("{}",df.collect().unwrap());
//     }
// }

use polars::prelude::*;

fn main() {
    let instant = std::time::Instant::now();
    let dir = std::fs::read_dir("data").unwrap();
    let lf = dir
        .into_iter()
        .filter(|x| {
            let x = x.as_ref().unwrap();
            let x = x.file_name();
            let x = x.to_str().unwrap();
            x.starts_with("2020-01")
        })
        .map(|x| {
            let x = x.unwrap();
            let x = x.path();
            println!("{:?}", x);
            let df = ParquetReader::new(std::fs::File::open(x).unwrap())
                .finish()
                .unwrap();
            let lf = df.lazy();
            let lf0 = lf
                .clone()
                .filter(col("sale_time").dt().hour().eq(lit(12)))
                .group_by([col("dan_wei"), col("saler")])
                .agg([col("sale_num").sum()])
                .group_by([col("dan_wei")])
                .agg([col("sale_num").mean()]);
            let lf1 = lf
                .clone()
                .filter(col("sale_time").dt().hour().eq(lit(12)))
                .group_by([col("dan_wei"), col("sale_good")])
                .agg([col("saler").n_unique()])
                .group_by([col("dan_wei")])
                .agg([col("saler").mean()]);
            let lf2 = lf
                .group_by([col("dan_wei"), col("sale_time").dt().hour()])
                .agg([col("sale_num").sum()])
                .group_by([col("dan_wei")])
                .agg([col("sale_num").mean()]);
            let lf = lf0
                .join(
                    lf1,
                    [col("dan_wei")],
                    [col("dan_wei")],
                    JoinArgs::new(JoinType::Inner),
                )
                .join(
                    lf2,
                    [col("dan_wei")],
                    [col("dan_wei")],
                    JoinArgs::new(JoinType::Inner),
                );
            let lf = lf.collect().unwrap();
            lf.lazy()
        })
        .collect::<Vec<_>>();
    let lf = concat(lf, UnionArgs::default())
        .unwrap()
        .group_by([col("dan_wei")])
        .agg([col("*").mean()])
        .collect()
        .unwrap();

    // let reader=std::fs::File::open("../data/2022-01-01.par").unwrap();
    // let df=ParquetReader::new(reader).finish().unwrap();
    print!("{},{}", lf, instant.elapsed().as_secs_f32());
}


